package com.ztrue.framework.disruptor;

import com.lmax.disruptor.EventFactory;
import com.ztrue.framework.exception.InvalidOperationException;
import com.ztrue.framework.infrastructure.IClone;
import com.ztrue.framework.infrastructure.IDistribute;
import com.ztrue.common.utilities.Assert;

import java.util.function.Consumer;

/**
 * 表示一个消息队列池
 *
 * @author zhangw
 * @date 2016/11/30 23:04
 * @version: 1.0
 */
public class MessageQueuePool<M extends IDistribute & IClone> implements IDistributeMessageQueue<M> {
    private IMessageQueue<M>[] messageQueue;
    private int poolSize;

    /**
     * 初始化消息队列池
     *
     * @param poolSize       池大小
     * @param bufferSize     缓冲区大小
     * @param threadNumbers  线程数量
     * @param messageHandler 消息处理器
     * @param eventFactory   事件工厂
     */
    public MessageQueuePool(int poolSize, int bufferSize, int threadNumbers, MessageHandler<M> messageHandler, EventFactory<M> eventFactory) {
        Assert.positiveInteger(poolSize, new InvalidOperationException("初始化队列数量异常"));
        Assert.positiveInteger(bufferSize, new InvalidOperationException("缓冲区大小数量异常"));
        if (threadNumbers < 2) {
            throw new InvalidOperationException("线程数量异常");
        }
        this.poolSize = poolSize;
        this.messageQueue = new MessageQueue[poolSize];
        for (int i = 0; i < poolSize; i++) {
            messageQueue[i] = MessageQueueFactory.make(messageHandler, bufferSize, threadNumbers, eventFactory);
        }
    }

    /**
     * 发送消息
     *
     * @param consumer
     * @param routingKey
     */
    @Override
    public void send(Consumer<M> consumer, String routingKey) {
        int n = routingKey.hashCode();
        messageQueue[((n ^ (n >> 31)) - (n >> 31)) % poolSize].send(consumer);
    }
}